package org.example;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.example.flatmap.MyFlatMap;
import org.example.sink.RedisExampleMapper;

import java.util.Properties;

/**
 * flink demo
 *
 * @author wlj
 */
public class KafkaSourceApplication {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 支持三种时间:处理时间(Processing Time),事件时间(Event Time),摄入时间(Ingestion time)
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        // 心跳检测
        env.enableCheckpointing(1000);
        // EXACTLY_ONCE,
        // AT_LEAST_ONCE;
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        Properties properties = new Properties();
        // 自行替换
        properties.setProperty("bootstrap.servers", "risk.flink.kafka.server");
        properties.setProperty("group.id", "risk.flink.kafka.group");

        DataStream<String> dataStream = env
                .addSource(new FlinkKafkaConsumer<>("risk.flink.kafka.topic", new SimpleStringSchema(), properties));

        // redis的配置
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("risk.flink.redis.host").setPassword("risk.flink.redis.password").setPort(6379).build();

        // FlatMap -> Sink
        dataStream.flatMap(new MyFlatMap()).addSink(new RedisSink<>(conf, new RedisExampleMapper()));

        env.execute("kafka");
    }
}
